package org.zhadoop.stream.cql;

import java.util.List;
import java.util.TreeMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.stream.api.Application;
import org.zhadoop.stream.api.UserFunction;
import org.zhadoop.stream.api.streams.Schema;
import org.zhadoop.stream.cql.builder.BuilderUtils;
import org.zhadoop.stream.cql.executor.FunctionRegistry;
import org.zhadoop.stream.cql.semanticanalyzer.parser.context.CreateOperatorContext;
import org.zhadoop.stream.cql.semanticanalyzer.parser.context.ParseContext;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

/**
 * driver 类中保存的临时数据
 * 便于初始化和清理
 * 
 * @author zwd
 *
 */
public class DriverContext {
	private static final Logger LOG = LoggerFactory.getLogger(DriverContext.class);
	
	private static final String FILE_PREFIX = "file://";
	
	/**
	 * 为每个线程生成一个命名空间，这样，流名称之类，就可以保持一致
	 */
	private static ThreadLocal<BuilderUtils> builderNameSpace = new ThreadLocal<BuilderUtils>();
	
	/**
	 * 客户端session内部可以使用的所有函数
	 */
	private static ThreadLocal<FunctionRegistry> functions = new ThreadLocal<FunctionRegistry>();
	
	/**
	 * 记录所有用户提交的CQL语句以及命令，只限于对系统会做出改变的命令
	 */
	private List<String> cqls = Lists.newArrayList();
	
	/**
	 * 存放所有CQL的解析AST树
	 * 同cqls，只限于对系统做出改变的命令
	 */
	private List<ParseContext> parseContexts = Lists.newArrayList();
	
	/**
	 * 流名称、schema信息
	 */
	private List<Schema> schemas = Lists.newArrayList();
	
	/**
	 * 用户下发的配置命令
	 * 因为最终要生成XML，为了避免xml中包含很多Streaming-site的配置属性
	 * 所以这里仅仅存放用户自定义的配置属性
	 */
	private TreeMap<String, String> userConfs = Maps.newTreeMap();
	
	/**
	 * 自定义的函数
	 * key:函数的名称
	 * value:函数信息
	 */
	private TreeMap<String,UserFunction> userDefinedFunctions = Maps.newTreeMap();
	 
	/**
	 * 自定义算子的名称
	 * key:算子名称
	 * value:自定义算子信息
	 */
	private TreeMap<String,CreateOperatorContext> userOperators = Maps.newTreeMap();
	
	/**
	 * 用户自定义的文件
	 */
	private List<String> userFiles = Lists.newArrayList();
	
	/**
	 * 已经编译成功或者load成功的app
	 */
	private Application app;
	
	/**
	 * CQL 命令执行结果
	 */
	private CQLResult queryResult;
	
	public DriverContext(){
		builderNameSpace.set(new BuilderUtils());
		functions.set(new FunctionRegistry());
	}

	public static ThreadLocal<FunctionRegistry> getFunctions() {
		return functions;
	}
	
	public static ThreadLocal<BuilderUtils> getBuilderNameSpace() {
		return builderNameSpace;
	}

	
	public void clean() {
		unRegisterJars();
		this.app = null;
		this.parseContexts.clear();
		this.cqls.clear();
		this.queryResult = null;
		this.schemas.clear();
		this.userConfs.clear();
		this.userDefinedFunctions.clear();
		this.userOperators.clear();
		builderNameSpace.remove();
		builderNameSpace.set(new BuilderUtils());
		functions.remove();
		functions.set(new FunctionRegistry());
	}
	
	
	private void unRegisterJars() {
		// TODO Auto-generated method stub
		
	}
	
	public List<String> getCqls() {
		return cqls;
	}

	public void setCqls(List<String> cqls) {
		this.cqls = cqls;
	}

	public Application getApp() {
		return app;
	}

	public void setApp(Application app) {
		this.app = app;
	}

	public void addParseContext(ParseContext parseContext){
		parseContexts.add(parseContext);
	}

	public List<ParseContext> getParseContexts() {
		return parseContexts;
	}

	public List<Schema> getSchemas(){
		return schemas;
	}
	
	public void addSchema(Schema schema){
		schemas.add(schema);
	}
	
	/**
	 * 用户下发自定义参数
	 * @param key
	 * @param value
	 */
	public void addConf(String key,String value){
		if(value == null){
			userConfs.remove(key);
		}else{
			userConfs.put(key, value);
		}
	}
	
	/**
	 * 添加命令
	 * @param cql
	 */
	public void addCQLs(String cql){
		cqls.add(cql);
	}
	/**
	 * 清空应用程序
	 */
	public void cleanApp() {
		app = null;
	}

	public CQLResult getQueryResult() {
		return queryResult;
	}

	public void setQueryResult(CQLResult queryResult) {
		this.queryResult = queryResult;
	}
	
	public TreeMap<String, String> getUserConfs(){
		return userConfs;
	}
}
	

	
